#include "FIFOQueueTest.h"
#include <atomic>
#include <cassert>
#include <cstddef>
#include <cstdint>
#include <cstdio>
#include <cstdlib>
#include <cstring>
#include <memory>
#include <sstream>
#include <unistd.h>
#include "PlatLogger.h"
#include "Util.h"
#include "XnccEventfd.h"
#include "XnccType.h"
namespace xncc {
namespace msg {
    using namespace xncc::shared;
    using namespace xncc::types;

#define aligned_queuelen_t __attribute__((aligned(64))) queue_size_t

    class FIFOQueueTestElementTest
    {
      public:
        FIFOQueueTestElementTest(queue_size_t size) : MASK(size - 1)
        {
            if (!CommonBusi::is_power_of_2(size)) {
                size = static_cast<queue_size_t>(CommonBusi::roundup_pow_of_two(size));
            }
            if (unlikely(size == 0)) {
                throw shared::system_exception("cant not alloc memory");
            }
            in           = 0;
            out          = 0;
            currReadLen  = 0;
            currWriteLen = 0;
            currentFinal = size;
            data         = reinterpret_cast<char*>(malloc(size));
            assert(data != nullptr);
        }
        ~FIFOQueueTestElementTest() { free(data); }

        inline queue_size_t capacity() const { return MASK + 1; }

        char* getWriteBuffer(queue_size_t len)
        {
            // currWriteLen   = len;
            // queue_size_t l = unused();
            // if (len > l) {
            //     return nullptr;  // 剩余空间不足
            // }
            // const queue_size_t size = MASK + 1;
            // queue_size_t       off  = in;
            // off &= MASK;
            // l = std::min(len, size - off);  // 后半段剩余的内存长度size - off
            // if (likely(len == l)) {
            //     return data + off;
            // }
            // const queue_size_t available_len = unused() - (size - off);  // 前半部分的剩余
            // if (available_len >= len) {
            //     currWriteLen = len + (size - off);  // 真实的写数据的长度 后半段空余的部分算做写了
            //     currentFinal = off;                 // 设置一个终点标记 给读使用  写不需要关心
            //     return data;
            // }
            const auto read_index  = static_cast<queue_size_t>(out & MASK);
            const auto write_index = static_cast<queue_size_t>(in & MASK);
            if (in >= out) {
                if (len > unused()) {
                    return nullptr;
                }
                if (capacity() - write_index >= len) {
                    currWriteLen = len;
                    if (write_index + len > currentFinal) {
                        currentFinal = capacity();
                    }
                    return data + write_index;
                }
                else {
                    if (len <= read_index) {
                        currWriteLen = len + static_cast<queue_size_t>(capacity() - write_index);
                        currentFinal = write_index;
                        return data;
                    }
                    else {
                        return nullptr;
                    }
                }
            }
            else {
                fprintf(stderr, "what happened ? memory error ?\n");
                // if (read_index - write_index >= len) {
                //     currWriteLen = len;
                //     if (write_index + len > currentFinal) {
                //         currentFinal = capacity();
                //     }
                //     return data + write_index;
                // }
                // else {
                //     return nullptr;
                // }
            }

            return nullptr;
        }

        void finishWrite()
        {
            smp_wmb();
            const auto result = static_cast<queue_size_t>(in + currWriteLen);
            in                = result;
        }

        char* getReadBuffer(queue_size_t len)
        {
            // currReadLen          = len;
            // const queue_size_t l = in - out;
            // if (len > l) {
            //     return nullptr;  // 没有那么多的数据可以读
            // }
            // const queue_size_t size = MASK + 1;
            // queue_size_t       off  = out;
            // off &= MASK;
            // const auto finial = currentFinal;
            // if (likely(off < finial)) {
            //     return data + off;
            // }
            // else if (off == finial) {
            //     currReadLen = len + (size - off);
            //     return data;
            // }
            // // // 读到这一轮的终点了  需要切换
            // // const queue_size_t available_len = in - out - (size - off);
            // // if (available_len >= len) {
            // //     currReadLen = len + (size - off);
            // //     return data;
            // // }
            const auto read_index  = out & MASK;
            const auto write_index = in & MASK;
            if (out <= in) {
                if (in - out >= len) {
                    if (read_index == currentFinal) {
                        if (write_index >= len) {
                            currReadLen = len + static_cast<queue_size_t>(capacity() - read_index);
                            return data;
                        }
                        else {
                            fprintf(stderr, "read  what happened ? %s \n", dump().c_str());
                        }
                    }
                    else {
                        currReadLen = len;
                        return data + read_index;
                    }
                }
                else {
                    return nullptr;
                }
            }
            else {
                fprintf(stderr, "read  what happened ? memory error ?\n");
                // if (read_index < currentFinal) {
                //     currReadLen = len;
                //     return data + read_index;
                // }
                // else if (read_index == currentFinal) {
                //     if (write_index >= len) {
                //         fprintf(stderr, "raraley happened dump:%s\n", dump().c_str());
                //         currReadLen = len + (capacity() - read_index);
                //         return data;
                //     }
                // }
                // else {
                //     fprintf(stderr, "what happened ? memory error ?\n");
                // }
            }

            return nullptr;
        }

        void finishRead()
        {
            smp_rmb();
            const auto result = static_cast<queue_size_t>(out + currReadLen);
            out               = result;
        }

        std::string dump()
        {
            std::ostringstream oss;
            oss << "in:" << in << " out:" << out << " in_off:" << (in & MASK) << " out_off:" << (out & MASK)
                << " size:" << capacity() << " currReadLen:" << currReadLen << " currWriteLen:" << currWriteLen
                << " currentFinal:" << currentFinal;
            return oss.str();
        }

      private:
        inline queue_size_t unused() { return static_cast<queue_size_t>((MASK + 1) - (in - out)); }

      private:
        const aligned_queuelen_t    MASK;
        volatile aligned_queuelen_t out;
        volatile aligned_queuelen_t currentFinal;
        volatile aligned_queuelen_t currReadLen;
        volatile aligned_queuelen_t in;
        volatile aligned_queuelen_t currWriteLen;
        char*                       data;
    };

    static std::atomic<types::queue_id_t> INIT_QUEUE_ID(1);
    FIFOQueueTest::FIFOQueueTest(queue_size_t size, bool autoExpand)
        : autoExpand_(autoExpand), lastReadIndex_(0), isNeedNotify_(false), queueId_(INIT_QUEUE_ID.fetch_add(1))
    {
        bufferList_[0] = std::make_shared<FIFOQueueTestElementTest>(size);
        currReadPtr_   = bufferList_[0];
        currWritePtr_  = bufferList_[0];
    }
    void FIFOQueueTest::setNeedNotify()
    {
        notifyfd_     = (std::make_shared<sys::EventFd>());
        isNeedNotify_ = true;
    }

    void FIFOQueueTest::updateCurrentWritePtr()
    {
        // 写指针的改变 以及使用都是在写线程操作，因此不存在多线程竞争问题
        std::lock_guard<std::mutex> lk(mutex_);
        int                         maxIndex = bufferList_.rbegin()->first;
        std::size_t                 oldCap   = bufferList_.rbegin()->second->capacity();
        maxIndex++;
        auto ptr              = std::make_shared<FIFOQueueTestElementTest>(oldCap * 2);  // 扩容
        bufferList_[maxIndex] = ptr;
        currWritePtr_         = ptr;
    }

    void FIFOQueueTest::updateCurrentReadPtr()
    {
        // 读指针的改变 以及使用都是在读线程操作，因此不存在多线程竞争问题
        std::lock_guard<std::mutex> lk(mutex_);
        int                         originIndex = lastReadIndex_;
        lastReadIndex_++;
        auto iter = bufferList_.find(lastReadIndex_);
        if (iter != bufferList_.end()) {
            currReadPtr_ = iter->second;
            bufferList_.erase(originIndex);
        }
        else {
            TRACE_ERROR << "what happened " << bufferList_.size() << " " << originIndex << " " << lastReadIndex_
                        << " dump read buffer " << currReadPtr_->dump();
            lastReadIndex_--;
        }
    }

    std::string FIFOQueueTest::dump() const
    {
        std::string ret;
        for (auto iter = bufferList_.begin(); iter != bufferList_.end(); iter++) {
            std::stringstream ss;
            ss << "index:" << iter->first << " " << iter->second->dump() << "  ";
            ret += ss.str();
        }
        return ret;
    }

    char* FIFOQueueTest::tryGetWriteBuffer(queue_size_t len) { return currWritePtr_->getWriteBuffer(len); }

    char* FIFOQueueTest::getWriteBuffer(queue_size_t len)
    {
        char* buf = currWritePtr_->getWriteBuffer(len);
        if (buf == nullptr) {  // buffer full failed
            if (autoExpand_) {
                updateCurrentWritePtr();                   // expand
                buf = currWritePtr_->getWriteBuffer(len);  // 只扩容一次
            }
        }
        return buf;
    }
    void FIFOQueueTest::finishWrite()
    {
        currWritePtr_->finishWrite();
        if (isNeedNotify_) {
            uint64_t one = 1;
            auto     n   = notifyfd_->write(one);
            if (n != sizeof one) {
                SYSTEM_ERROR << "event write failed fd:"_s << notifyfd_->Fd() << " queue id:"_s << queueId_;
            }
        }
    }
    char* FIFOQueueTest::getReadBuffer(queue_size_t len)
    {
        char* buf = currReadPtr_->getReadBuffer(len);
        if (autoExpand_) {
            if (buf == nullptr) {
                bool needUpdate = true;
                if (!isNeedNotify_) {  // 不需要通知就进行读操作的read
                    // 可能没有数据的时候就开始读了，这个时候自动扩容的情况下 没有扩容情况下 不更新读指针
                    std::size_t listSize;
                    {
                        std::lock_guard<std::mutex> lk(mutex_);
                        listSize = bufferList_.size();
                    }
                    if (listSize == 1) {
                        needUpdate = false;
                    }
                }
                if (needUpdate) {
                    updateCurrentReadPtr();
                    buf = currReadPtr_->getReadBuffer(len);
                }
            }
        }
        return buf;
    }
    void FIFOQueueTest::finishRead() { currReadPtr_->finishRead(); }
}  // namespace msg
}  // namespace xncc